Spark常用优化方法

2023-05-19 00:19| 来源: 网络整理| 查看: 265

一、前言1.为什么要优化？因为你的资源有限、更快速的跑完任务、防止不稳定因素导致的任务失败。2.怎样做优化？通常查看spark的web UI，或者查看运行中的logs3.做哪方面的优化？ spark 应用程序 80% 的优化，都是集中在三个地方：内存，磁盘io，网络io二、调优详情1.spark-submit命令中作为参数设置

资源参数设置的不合理，可能会导致没有充分利用集群资源，作业运行会极其缓慢；或者设置的资源过大，队列没有足够的资源来提供，进而导致各种异常。

--num-executors //该参数用于设置Spark作业总共要用多少个Executor进程来执行 --executor-memory //该参数用于设置每个Executor进程的内存。Executor内存的大小，很多时候直接决定了Spark作业的性能，而且跟常见的JVM OOM异常，也有直接的关联。 //每个Executor进程的内存设置4G~8G较为合适。num-executors乘以executor-memory，就代表了你的Spark作业申请到的总内存量（也就是所有Executor进程的内存总和） --executor-cores //该参数用于设置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。同样建议，如果是跟他人共享这个队列，那么num-executors * executor-cores不要超过队列总CPU core的1/3~1/2左右比较合适 -- driver-memory //设置Driver进程的内存，默认1G即可 -- spark.default.parallelism //该参数用于设置每个stage的默认task数量。这个参数极为重要。【如果不去设置这个参数，那么此时就会导致Spark自己根据底层HDFS的block数量来设置task的数量，默认是一个HDFS block对应一个task。task数量偏少的话，就会导致你前面设置好的Executor的参数都前功尽弃。】 --spark.storage.memoryFraction //用于设置RDD持久化数据在Executor内存中能占的比例，默认是0.6 --spark.shuffle.memoryFraction //设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认是0.2./bin/spark-submit \ --master yarn-cluster \ --num-executors 100 \ --executor-memory 6G \ --executor-cores 4 \ --driver-memory 1G \ --conf spark.default.parallelism=1000 \ --conf spark.storage.memoryFraction=0.5 \ --conf spark.shuffle.memoryFraction=0.3 \1.提高并行度parallelism

如果设置了new SparkConf().set(“spark.default.parallelism”,“5”)，所有的RDDpartition都会被设置为 5个，也就是每个RDD的数据都会被分为5分，每个partition都会启动一个task来进行计算，对于所有的算子操作，都只会使用5个task在集群中运行。所以在这个时候，集群中有10个cpucore，也只会使用5个来进行task，剩余空闲。造成资源浪费。

面对10个core，我们可以设置10个甚至20、30个task，因为task之间的执行顺序和时间是不一样的，正好10个也会造成浪费。官方建议并行度设置为core数的2~3倍，可以最高效率使用资源。

2. repartition and coalesce3.使用Kryo序列化机制

spark默认使用了java自身提供的序列化机制，基于ObjectInputStream和OBjectOutputStream的序列化机制，但是性能比较差，速度慢，序列化之后占用空间依旧高

spark还提供了另外一种序列化机制——Kryo序列化机制，快，结果集小10倍，缺点是有些类型及时实现了Seriralizable接口，也不一定能被序列化。如果想要达到最好的性能，Kryo要求在Spark中对所有需要序列化的类型进行注册

4.优化数据结构（属于代码内优化）5.对于多次使用的RDD进行持久化或者Checkpoint6.设置广播共享数据

使用Broadcast广播，让其在每个节点中一个副本，而不是每个task一个副本。减少节点上的内存占用。 Broadcast广播节点上使用该数据的时候不需要调用RDD，而是调用broadcastConf广播副本，就可以节省内存和网络IO

7. 数据本地化（spark的内部优化机制）

数据本地化处理机制（基于数据距离代码的距离）情况由好到坏：

（1）PROCESS_LOCAL:数据和计算它的代码在同一个JVM进程中

（2）NODE_LOCAL:数据和代码在同一节点上，但是不在一个进程中，比如在不容的executor进程汇总，或者数据在HDFS文件的block中

（3）NO_PREF:数据从哪里过来，最终的性能都是一样的

（4）RACK_LOCAL:数据和代码在同一个机架上

（5）ANY:数据可能在任意地方，比如其他网络环境或者其他机架上

8.reduceByKey和groupByKey

一般情况下，reduceByKey的操作都是可以使用groupByKey().map()来进行替代操作的。但是 groupByKey不会进行本地聚合，原封不动的将ShuffleMapTask的输出拉渠道ResultTask的内存中。因为reduceByKey首先会在map端进行本地的combine，可以大大减要传输到reduce的数据量，减少网络IO，只有在reduceByKey解决不了类似问题的时候才会使用groupByKey().map()来进行代替。

一、Spark性能优化：开发调优篇二、Spark性能优化：资源调优篇三、Spark性能优化：数据倾斜调优四、Spark性能优化：shuffle调优

【本文地址】

公司简介

联系我们